package com.zhike.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.zhike.bean.TripBrandTypeBean;
import com.zhike.function.DimAsyncFunction;
import com.zhike.utils.DateFormatUtil;
import com.zhike.utils.DorisUtils;
import com.zhike.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhike
 * @CreateTime: 2024/1/29 9:34
 * @Description: 计算各车型，品牌各窗口的速度，行驶的里程，急加减次数并写入doris
 */
public class DwsTripBrandTypeWindow {
    public static void main(String[] args) {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.setParallelism(2);

        //2.读取car_electricity_log的数据创建流并转换为JavaBean对象
        SingleOutputStreamOperator<TripBrandTypeBean> electricityDS = env.fromSource(KafkaUtils.getKafkaSource("car_electricity_log", "dws_trip_brand_type_window_1"),
                        WatermarkStrategy.noWatermarks(), "ele_source")
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, TripBrandTypeBean>() {
                    @Override
                    public TripBrandTypeBean map(JSONObject value) throws Exception {
                        return getTripBrandTypeBean(value);
                    }

                });

        //3.读取car_hybrid_log的数据创建流并转换为JavaBean对象
        SingleOutputStreamOperator<TripBrandTypeBean> hybridDS = env.fromSource(KafkaUtils.getKafkaSource("car_hybrid_log", "dws_trip_brand_type_window_2"),
                        WatermarkStrategy.noWatermarks(), "hybrid_source")
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, TripBrandTypeBean>() {
                    @Override
                    public TripBrandTypeBean map(JSONObject value) throws Exception {
                        return getTripBrandTypeBean(value);
                    }

                });

        //4.读取car_fuel_log的数据创建流并转换为JavaBean对象
        SingleOutputStreamOperator<TripBrandTypeBean> fuelDS = env.fromSource(KafkaUtils.getKafkaSource("car_fuel_log", "dws_trip_brand_type_window_3"),
                        WatermarkStrategy.noWatermarks(), "fuel_source")
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, TripBrandTypeBean>() {
                    @Override
                    public TripBrandTypeBean map(JSONObject value) throws Exception {
                        return getTripBrandTypeBean(value);
                    }

                });

        //5.合并3个流
        DataStream<TripBrandTypeBean> unionDS = electricityDS.union(hybridDS, fuelDS);

        //6.设置水位线并按照vin分组
        KeyedStream<TripBrandTypeBean, String> keyedVinDS = unionDS.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<TripBrandTypeBean>forBoundedOutOfOrderness(Duration.ofSeconds(30L))
                                .withTimestampAssigner(new SerializableTimestampAssigner<TripBrandTypeBean>() {
                                    @Override
                                    public long extractTimestamp(TripBrandTypeBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }))
                .keyBy(new KeySelector<TripBrandTypeBean, String>() {
                    @Override
                    public String getKey(TripBrandTypeBean value) throws Exception {
                        return value.getVin();
                    }
                });

        //7.通过状态编程计算急加减次数和当前行驶里程
        SingleOutputStreamOperator<TripBrandTypeBean> velocityStateDS = keyedVinDS.map(new RichMapFunction<TripBrandTypeBean, TripBrandTypeBean>() {

            private ValueState<Double> velocityState;
            private ValueState<Double> mileageState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //获取状态
                velocityState = getRuntimeContext().getState(new ValueStateDescriptor<>("velocity_state", Double.class));
                mileageState = getRuntimeContext().getState(new ValueStateDescriptor<>("mileage_state", Double.class));
            }

            @Override
            public TripBrandTypeBean map(TripBrandTypeBean value) throws Exception {
                //获取当前速度
                Double totalSpeed = value.getTotalSpeed();
                //获取状态中的值
                Double velocityPre = velocityState.value();
                if (velocityPre == null) {
                    velocityState.update(totalSpeed);
                } else {
                    if (Math.abs(totalSpeed - velocityPre) > 70) {
                        value.setDangerCount(value.getDangerCount() + 1);
                        velocityState.update(totalSpeed);
                    }
                    velocityState.update(totalSpeed);
                }

                //获取当前里程
                Double totalMileage = value.getTotalMileage();
                //获取里程状态中的值
                Double mileagePre = mileageState.value();
                if (mileagePre == null) {
                    value.setTotalMileage(0.0);
                    mileageState.update(totalMileage);
                }else {
                    value.setTotalMileage(Math.abs(totalMileage-mileagePre));
                    mileageState.update(totalMileage);
                }
                return value;
            }
        });


        //8.关联car_info维表，补充车型，品牌维度
        SingleOutputStreamOperator<TripBrandTypeBean> rishDS = AsyncDataStream.unorderedWait(velocityStateDS,
                new DimAsyncFunction<>("DIM_CAR_INFO") {
                    @Override
                    public String getKey(TripBrandTypeBean input) {
                        return input.getVin();
                    }

                    @Override
                    public void join(TripBrandTypeBean input, JSONObject dimInfo) {
                        input.setCarType(dimInfo.getString("TYPE"));
                        input.setBrand(dimInfo.getString("TRADEMARK"));
                    }
                }, 100L, TimeUnit.SECONDS);


        //9.按照车型，品牌分组
        KeyedStream<TripBrandTypeBean, Tuple2<String, String>> keyedCBDS = rishDS.keyBy(
                new KeySelector<TripBrandTypeBean, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(TripBrandTypeBean value) throws Exception {
                        return Tuple2.of(value.getCarType(), value.getBrand());
                    }
                }
        );


        //10.开窗聚合
        SingleOutputStreamOperator<TripBrandTypeBean> reduceDS = keyedCBDS.window(TumblingEventTimeWindows.of(Time.seconds(60L)))
                .reduce(new ReduceFunction<TripBrandTypeBean>() {
                    @Override
                    public TripBrandTypeBean reduce(TripBrandTypeBean value1, TripBrandTypeBean value2) throws Exception {
                        value1.setTotalMileage(value1.getTotalMileage() + value2.getTotalMileage());
                        value1.setTotalSpeed(value1.getTotalSpeed() + value2.getTotalSpeed());
                        value1.setRunningCount(value1.getRunningCount() + value2.getRunningCount());
                        value1.setDangerCount(value1.getDangerCount() + value2.getDangerCount());
                        return value1;
                    }
                }, new WindowFunction<TripBrandTypeBean, TripBrandTypeBean, Tuple2<String, String>, TimeWindow>() {

                    @Override
                    public void apply(Tuple2<String, String> stringStringTuple2, TimeWindow window, Iterable<TripBrandTypeBean> input, Collector<TripBrandTypeBean> out) throws Exception {
                        TripBrandTypeBean bean = input.iterator().next();
                        bean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        bean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        bean.setTs(System.currentTimeMillis());
                        bean.setWindowDate(DateFormatUtil.toDate(window.getStart()));
                        out.collect(bean);
                    }
                });


        //11.将结果写入Doris
        SingleOutputStreamOperator<String> resultDS = reduceDS.map(new MapFunction<TripBrandTypeBean, String>() {
            @Override
            public String map(TripBrandTypeBean value) throws Exception {
                SerializeConfig serializeConfig = new SerializeConfig();
                serializeConfig.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                return JSONObject.toJSONString(value, serializeConfig, SerializerFeature.WriteMapNullValue);
            }
        });

        resultDS.print();
        resultDS.sinkTo(DorisUtils.getDorisSink("dws_trip_bt_window"));


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }

    private static TripBrandTypeBean getTripBrandTypeBean(JSONObject value) {
        double velocity = value.getInteger("velocity") / 10.0;
        return TripBrandTypeBean.builder()
                .vin(value.getString("vin"))
                .totalMileage(value.getInteger("mileage") / 10.0)
                .totalSpeed(velocity)
                .runningCount(velocity > 0 ? 1 : 0)
                .dangerCount(0)
                .ts(value.getLong("timestamp"))
                .build();
    }
}
